package com.us.kafka;
import java.math.BigDecimal;
import java.util.Date;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

import com.alibaba.fastjson.JSONArray;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;  
import kafka.producer.ProducerConfig;  
import kafka.serializer.StringEncoder;
import com.alibaba.fastjson.JSONObject;

/**
 * 
 * @author yangyibo
 *
 */
public class KafkaProducer extends Thread{
	 private String topic;
	 private JSONObject jsonObject;
     
	    public KafkaProducer(String topic,JSONObject jsonObject){
	        super();  
	        this.topic = topic;
	        this.jsonObject = jsonObject;
	    }
	      
	      
	    @Override  
	    public void run() {  
	        Producer producer = createProducer();  
	        int i=1;
	        while(i<=1){
	        	i++;
                String code = UniqueCodeUntil.getOrderIdByUUId(16);
                String code2 = UniqueCodeUntil.getOrderIdByUUId(16);
                code = code+ code2.substring(10,15);
                System.out.println(code);
                jsonObject.put("businessKey",code);
                String codeq = UniqueCodeUntil.getOrderIdByUUId(16);
                int aa =(int)(1+Math.random()*(100-1+1));
                jsonObject.put("alarmTypeName", "测试"+ Integer.toString(aa));
                jsonObject.put("alarmCategoryName", "刑事案件"+codeq);
                int bb =(int)(1+Math.random()*(100-1+1));
                jsonObject.put("contents", "TTTTTTTTTTTTT"+ Integer.toString(bb));
//				jsonObject.put("speed",code.substring(code.length()-5,code.length()));
                System.out.println("jsonObject= " + "AAAA"+i);
	            producer.send(new KeyedMessage<Integer, String>(topic, jsonObject.toJSONString()));
//				System.out.println("插入第 "+ i +"    条警情信息,警情ode = " +code );
	            try {
	                TimeUnit.SECONDS.sleep(1);
	            } catch (InterruptedException e) {  
	                e.printStackTrace();  
	            }  
	        }  
	    }  
	  
		private Producer createProducer() {  
	        Properties properties = new Properties();
			//声明zk
	        properties.put("zookeeper.connect", KafkaConfig.zookeeper);
	        properties.put("serializer.class", StringEncoder.class.getName());
			// 声明kafka broker
	        properties.put("metadata.broker.list", KafkaConfig.metadata_broker_list);
            properties.put("acks", "all");
            properties.put("retries", 0);
            properties.put("batch.size", 16384);
            properties.put("linger.ms", 1);
            properties.put("buffer.memory", 33554432);
            properties.put("num.replica.fetchers", 1);
            properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            return new Producer<Integer, String>(new ProducerConfig(properties));
	     }  
	      
	      
		/**
		 * 
		 * @param topicName
		 */
	    public static void creatProducer(String topicName ,JSONObject json) {

	            //警情推送
                JSONArray jsa = new JSONArray();
                json.put("alarmTime", DateUtil.toDateString("yyyy/MM/dd HH:mm:ss",new Date()));
                json.put("origin", "110");
                json.put("levels", "1");
                json.put("aPerson", "AA");
                json.put("aPersonPhone", "13088888888");
                json.put("address", "黄鹤楼");
                json.put("longitude", "114.2562234");
                json.put("latitude", "30.6189490");
                json.put("createTime", DateUtil.toDateString("yyyy/MM/dd HH:mm:ss",new Date()));
                json.put("disposeStatus", "警情处理中");
                JSONObject json1 = new JSONObject();
                json1.put("policeName", "高茜");
                json1.put("policeId", "f779b30d-25d7-4a53-8c18-08b82e2b9199");
                json1.put("stationId", new BigDecimal("427100000000"));
                json1.put("deviceCatalog", "car");
            json1.put("stationName", "江汉区分局治安管理大队");
            JSONObject json2 = new JSONObject();
            json2.put("policeName", "高茜");
            json2.put("policeId", "f779b30d-25d7-4a53-8c18-08b82e2b9198");
            json2.put("stationId", new BigDecimal("427100000000"));
            json2.put("deviceCatalog", "car");
            json2.put("stationName", "江汉区分局治安管理大队");
            JSONObject json3 = new JSONObject();
            json3.put("policeName", "高茜");
            json3.put("policeId", "f779b30d-25d7-4a53-8c18-08b82e2b9198");
            json3.put("stationId", new BigDecimal("427100000000"));
            json3.put("deviceCatalog", "car");
            json3.put("stationName", "江汉区分局治安管理大队");
            JSONObject json4 = new JSONObject();
            json4.put("policeName", "高茜");
            json4.put("policeId", "f779b30d-25d7-4a53-8c18-08b82e2b9198");
            json4.put("stationId", new BigDecimal("427100000000"));
            json4.put("deviceCatalog", "car");
            json4.put("stationName", "江汉区分局治安管理大队");
            JSONObject json5 = new JSONObject();
            json5.put("policeName", "高茜");
            json5.put("policeId", "f779b30d-25d7-4a53-8c18-08b82e2b9198");
            json5.put("stationId", new BigDecimal("427100000000"));
            json5.put("deviceCatalog", "car");
            json5.put("stationName", "江汉区分局治安管理大队");


            json1.put("deviceMark", "鄂A01592");
            json2.put("deviceMark", "鄂A10544");
            json3.put("deviceMark", "鄂A02491");
            json4.put("deviceMark", "鄂A13753");

            jsa.add(json2);

            json.put("policeList", jsa);
	        new KafkaProducer(topicName,json).start();
	    }  
	       
	} 